package consumer

import (
	"context"
	"errors"
	"fmt"
	"github.com/Shopify/sarama"
	"log"
	"math/rand"
)

const (
	gTestPrefix = "test_consumer_"
)

var (
	defVersion = sarama.V0_10_2_2 //版本
)

type (
	HandleErrorFunc func(error)
	// 如果返回error，则这一条消费不会被mark消费成功
	HandleConsumerFunc func(data []byte) error
)

type Config struct {
	// broker的集群地址
	Brokers []string `json:"brokers"`
	// topic 的名称
	Topic string `json:"topic"`
	// 消费组名称
	Group string `json:"group":"group"`

	SASL struct {
		Enable   bool   `json:"enable"`
		User     string `json:"user"`
		Password string `json:"password"`
	} `json:"sasl"`
	// 是否从最老的开始消费
	Oldest bool `json:"oldest"`
}

type Consumer struct {
	ctx       context.Context
	group     sarama.ConsumerGroup
	cancel    context.CancelFunc
	saramaCfg *sarama.Config
	cfg       *Config

	// 消费处理失败方法
	errHandler HandleErrorFunc
	// 消费者处理方法
	consumerTopicHandle HandleConsumerFunc
}

func (c *Config) Check() bool {
	if len(c.Topic) == 0 || len(c.Brokers) == 0 {
		return false
	}
	if c.Group == "" {
		c.Group = fmt.Sprintf("%s%.8x", gTestPrefix, rand.Int63())
	}
	return true
}

func NewConsumerWithConfig(cfg *Config) (*Consumer, error) {
	if !cfg.Check() {
		return nil, errors.New("config error")
	}
	saramaCfg := sarama.NewConfig()
	{
		saramaCfg.Consumer.Return.Errors = true
		saramaCfg.Version = GetVersion()
		if cfg.Oldest {
			saramaCfg.Consumer.Offsets.Initial = sarama.OffsetOldest
		} else {
			saramaCfg.Consumer.Offsets.Initial = sarama.OffsetNewest
		}
		if cfg.SASL.Enable {
			saramaCfg.Net.SASL.Enable = cfg.SASL.Enable
			saramaCfg.Net.SASL.User = cfg.SASL.User
			saramaCfg.Net.SASL.Password = cfg.SASL.Password
		}
	}
	if err := saramaCfg.Validate(); err != nil {
		return nil, err
	}

	// Start with a client
	client, err := sarama.NewClient(cfg.Brokers, saramaCfg)
	if err != nil {
		return nil, err
	}

	// Start a new consumer group
	consumerGroup, err := sarama.NewConsumerGroupFromClient(cfg.Group, client)
	if err != nil {
		return nil, err
	}

	c := new(Consumer)
	c.cfg = cfg
	c.saramaCfg = saramaCfg
	c.group = consumerGroup
	c.ctx, c.cancel = context.WithCancel(context.Background())
	return c, nil
}

func GetVersion() sarama.KafkaVersion {
	return defVersion
}

func (c *Consumer) Close() error {
	log.Println("kafka consumer close")
	c.cancel()
	return c.group.Close()
}

type GroupHandler struct {
	consumer *Consumer
}

func (GroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (GroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (g GroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case msg, ok := <-claim.Messages():
			if !ok {
				return nil
			}
			log.Printf("Message topic:%q partition:%d offset:%d\n,", msg.Topic, msg.Partition, msg.Offset)
			if g.consumer != nil && g.consumer.consumerTopicHandle != nil {
				err := g.consumer.consumerTopicHandle(msg.Value)
				if err != nil {
					return err
				}
			} else {
				return errors.New("not set handler")
			}

			//标记消息已经被消费
			sess.MarkMessage(msg, "")
		}
	}
	return nil
}

func (c *Consumer) Run() {
	log.Println("kafka consumer running")
	// 消费数据
	go func() {
		topics := []string{c.cfg.Topic}
		handler := GroupHandler{c}

		for {
			if err := c.group.Consume(c.ctx, topics, handler); err != nil {
				if c.errHandler != nil {
					c.errHandler(err)
				}
			}

			if c.ctx.Err() != nil {
				return
			}
		}
	}()

	// 处理消费组错误
	go func() {
		for {
			select {
			case err, ok := <-c.group.Errors():
				{
					if !ok {
						return
					}
					if c.errHandler != nil {
						c.errHandler(err)
					}
				}
			}
		}
	}()
}

func (c *Consumer) HandlerTopic(handler HandleConsumerFunc) {
	c.consumerTopicHandle = handler
}

func (c Consumer) HandlerErr(e HandleErrorFunc) {
	c.errHandler = e
}
